// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: taekwon

#pragma once

#include <experimental/filesystem>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <google/protobuf/util/message_differencer.h>
#include <nlohmann/json.hpp>

#include "coin/proto/coin_order_gateway.pb.h"
#include "coin/proto/coin_request.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/date_util.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/log/proto_logger.h"
#include "orio/io/base.h"
#include "orio/record/topic.h"

namespace coin2::exchange::base::order_util {

namespace impl {

using coin::proto::OrderGatewayLog;

inline coin::proto::AccountBalance GetNoneZeroBalance(
    const coin::proto::AccountBalance& orig_acc_balance) {
  coin::proto::AccountBalance acc_balance;
  acc_balance.CopyFrom(orig_acc_balance);
  acc_balance.clear_each_balance();
  for (int i = 0; i < orig_acc_balance.each_balance_size(); i++) {
    const auto& orig_ccy_balance = orig_acc_balance.each_balance(i);
    if (orig_ccy_balance.total() != 0.) {
      acc_balance.add_each_balance()->CopyFrom(orig_ccy_balance);
    }
  }
  return acc_balance;
}

inline coin::proto::AccountPosition GetNoneZeroPosition(
    const coin::proto::AccountPosition& orig_acc_pos) {
  coin::proto::AccountPosition acc_pos;
  acc_pos.CopyFrom(orig_acc_pos);
  acc_pos.clear_each_position();
  acc_pos.clear_each_margin_usage();
  for (int i = 0; i < orig_acc_pos.each_position_size(); i++) {
    const auto& orig_prod_pos = orig_acc_pos.each_position(i);
    if (orig_prod_pos.net_position() != 0. || orig_prod_pos.net_position_contract() != 0. ||
        orig_prod_pos.long_position() != 0. || orig_prod_pos.short_position() != 0. ||
        orig_prod_pos.realized_pnl() != 0. || orig_prod_pos.unrealized_pnl() != 0.) {
      acc_pos.add_each_position()->CopyFrom(orig_prod_pos);
    }
  }
  for (int i = 0; i < orig_acc_pos.each_margin_usage_size(); i++) {
    const auto& margin_usage = orig_acc_pos.each_margin_usage(i);
    if (margin_usage.total_margin() != 0.) {
      acc_pos.add_each_margin_usage()->CopyFrom(margin_usage);
    }
  }
  return acc_pos;
}

inline coin::proto::CurrencyTransferBundle GetNoneZeroTransfer(
    const coin::proto::CurrencyTransferBundle& orig_ccy_transfer_bundle) {
  coin::proto::CurrencyTransferBundle ccy_transfer_bundle;
  ccy_transfer_bundle.CopyFrom(orig_ccy_transfer_bundle);
  ccy_transfer_bundle.clear_each_transfer();
  for (int i = 0; i < orig_ccy_transfer_bundle.each_transfer_size(); i++) {
    const auto& ccy_transfer = orig_ccy_transfer_bundle.each_transfer(i);
    if (ccy_transfer.qty() != 0.) {
      ccy_transfer_bundle.add_each_transfer()->CopyFrom(ccy_transfer);
    }
  }
  return ccy_transfer_bundle;
}

struct OrderLogPathComponent {
  // e.g. jang.Futures.Huobi.20191223-020741.1577066861418000000
  std::string owner;
  std::string market;
  std::string exchange;
  std::string log_time_str;
  int64_t log_timestamp{};

  std::string GetFilenameFromLogPathComponent() const {
    return fmt::format("{}.{}.{}.{}.{}", owner, market, exchange, log_time_str, log_timestamp);
  }

  static std::optional<OrderLogPathComponent> GetLogPathComponentFromFilename(
      const std::string& filename) {
    if (std::count(filename.begin(), filename.end(), '.') != 4 &&
        std::count(filename.begin(), filename.end(), '.') != 5) {
      return {};
    }

    OrderLogPathComponent ret;
    std::istringstream iss(filename);
    getline(iss, ret.owner, '.');
    getline(iss, ret.market, '.');
    getline(iss, ret.exchange, '.');
    getline(iss, ret.log_time_str, '.');
    std::string log_timestamp_str;
    getline(iss, log_timestamp_str, '.');
    ret.log_timestamp = stoll(log_timestamp_str);
    return ret;
  }
};

class OrderLogger : public BaseProtoLogger {
 public:
  using MessageDiff = google::protobuf::util::MessageDifferencer;

 public:
  OrderLogger(
      const coin::proto::AccountRequestProto& request,
      const std::string& log_root,
      const std::string& machine,
      const std::string& extension = "",
      bool use_slim_log_mode = false)
      : BaseProtoLogger(log_root, machine),
        request_(request),
        extension_(extension),
        use_slim_log_mode_(use_slim_log_mode) {
    SetInitCallbackFunc([this](int64_t ts) { this->WriteAccountRequest(true, ts); });
  }

  bool WriteAccountRequest(bool flush, int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::ACCOUNT_REQUEST);
    op_log_proto_.set_raw_msg_id(og_id_);
    *op_log_proto_.mutable_account_request() = request_;
    if (callback_) {
      callback_(op_log_proto_);
    }
    std::string buf;
    if (!op_log_proto_.SerializeToString(&buf)) {
      return false;
    }
    Write(buf, flush, timestamp);
    return true;
  }

  bool WritePosition(const coin::proto::AccountPosition& pos, bool flush, int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::POSITION);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    *op_log_proto_.mutable_account_request() = request_;
    if (as_account_logger_) {
      *op_log_proto_.mutable_position() = GetNoneZeroPosition(pos);
    } else {
      *op_log_proto_.mutable_position() = pos;
    }
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!MakeSlimForPositionLog(timestamp, pos, op_log_proto_.mutable_position())) {
      return false;
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteBalance(const coin::proto::AccountBalance& balance, bool flush, int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::BALANCE);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    *op_log_proto_.mutable_account_request() = request_;
    if (as_account_logger_) {
      *op_log_proto_.mutable_balance() = GetNoneZeroBalance(balance);
    } else {
      *op_log_proto_.mutable_balance() = balance;
    }
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!MakeSlimForBalanceLog(timestamp, balance, op_log_proto_.mutable_balance())) {
      return false;
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteTransfer(
      const coin::proto::CurrencyTransferBundle& transfer,
      bool flush,
      int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::TRANSFER);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    *op_log_proto_.mutable_account_request() = request_;
    *op_log_proto_.mutable_transfer() = transfer;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool
  WriteFundingFee(const coin::proto::FundingFeeBundle& funding_fee, bool flush, int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::FUNDING_FEE);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    *op_log_proto_.mutable_account_request() = request_;
    *op_log_proto_.mutable_funding_fee() = funding_fee;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteNftTokenInfo(
      const ::coin::proto::NftTokenInfoBundle& nft_token_info,
      bool flush,
      int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::NFT_TOKEN_INFO);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    *op_log_proto_.mutable_account_request() = request_;
    *op_log_proto_.mutable_nft_token_info() = nft_token_info;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteOrderEvent(const coin::proto::OrderEvent& order_event, bool flush, int64_t timestamp) {
    if (as_account_logger_) {
      if (order_event.type() != coin::proto::OrderEvent::ORDER_FILLED) {
        return false;
      }
    }

    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::ORDER_EVENT);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    *op_log_proto_.mutable_account_request() = request_;
    *op_log_proto_.mutable_event() = order_event;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);

    ::nlohmann::json msg = {
        {"compartment", "order"},
        {"component", "OrderLogger"},
        {"order_event", op_log_proto_.DebugString()},
    };
    SPDLOG_INFO(msg.dump());

    return true;
  }

  bool WritePrivateExchangeMessage(const std::string& raw_msg, bool flush, int64_t timestamp) {
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::PRIVATE_EXCHANGE_MESSAGE);
    op_log_proto_.set_raw_msg_id(raw_msg_id_);
    op_log_proto_.set_private_exchange_message(raw_msg);
    *op_log_proto_.mutable_account_request() = request_;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteRateLimitReportMessage(
      const coin::proto::RateLimitReportMessage& rate_limit_report_msg,
      bool flush,
      int64_t timestamp) {
    if (as_account_logger_) {
      return false;
    }
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::RATE_LIMIT_REPORT_MESSAGE);
    *op_log_proto_.mutable_rate_limit_report_message() = rate_limit_report_msg;
    *op_log_proto_.mutable_account_request() = request_;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteRiskEvent(const coin::proto::RiskEvent& risk_event, bool flush, int64_t timestamp) {
    if (as_account_logger_) {
      return false;
    }
    op_log_proto_.Clear();
    op_log_proto_.set_type(OrderGatewayLog::RISK_EVENT);
    *op_log_proto_.mutable_risk_event() = risk_event;
    *op_log_proto_.mutable_account_request() = request_;
    if (callback_) {
      callback_(op_log_proto_);
    }
    if (!op_log_proto_.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  std::string GetEnsuredLogPath(int64_t timestamp) const final {
    std::string log_filename =
        OrderLogPathComponent{
            request_.owner(),
            request_.market_type(),
            request_.exchange(),
            GetDateTime(timestamp),
            timestamp}
            .GetFilenameFromLogPathComponent();
    std::experimental::filesystem::path log_dir(log_root_);
    log_dir /= machine_;
    log_dir /= coin2::exchange::base::symbology::ToDateStr(log_timestamp());
    EnsureDirExists(log_dir.native());
    std::experimental::filesystem::path log_path(log_dir);
    log_path /= log_filename;
    return log_path.native() + extension_;
  }

  const coin::proto::AccountRequestProto& request() const { return request_; }
  void SetCallback(const std::function<void(const coin::proto::OrderGatewayLog& log)>& callback) {
    callback_ = callback;
  }

  void set_raw_msg_id(int64_t raw_msg_id) { raw_msg_id_ = raw_msg_id; }
  void set_og_id(int64_t og_id) { og_id_ = og_id; }
  void set_as_account_logger(bool v) { as_account_logger_ = v; }

 private:
  bool MakeSlimForBalanceLog(
      int64_t timestamp,
      const coin::proto::AccountBalance& bal,
      coin::proto::AccountBalance* proto_bal) {
    // TODO(daniel): improve performance
    if (!use_slim_log_mode_) {
      return true;
    }
    bool has_bal_update = false;
    if (last_log_bal_snapshot_ts_ > 0 && last_log_bal_snapshot_ts_ + 10'000'000'000LL > timestamp) {
      // write update
      proto_bal->mutable_each_balance()->Clear();
      coin::proto::AccountBalance new_bal;
      new_bal.set_exchange(bal.exchange());
      new_bal.set_market_type(bal.market_type());
      for (const auto& ele : bal.each_balance()) {
        const auto& currency = ele.currency();
        if (curr_bal_map_.count(currency) &&
            MessageDiff::ApproximatelyEquals(curr_bal_map_[currency], ele)) {
          continue;
        }
        curr_bal_map_[currency] = ele;
        new_bal.add_each_balance()->CopyFrom(ele);
        has_bal_update = true;
      }
      if (has_bal_update) {
        *proto_bal = new_bal;
      }
    } else {
      // write snapshot
      for (const auto& ele : bal.each_balance()) {
        const auto& currency = ele.currency();
        curr_bal_map_[currency] = ele;
      }
      last_log_bal_snapshot_ts_ = timestamp;
      has_bal_update = true;
    }
    return has_bal_update;
  }

  bool MakeSlimForPositionLog(
      int64_t timestamp,
      const coin::proto::AccountPosition& pos,
      coin::proto::AccountPosition* proto_pos) {
    // TODO(daniel): improve performance
    if (!use_slim_log_mode_) {
      return true;
    }
    bool has_pos_update = false;
    if (last_log_pos_snapshot_ts_ > 0 && last_log_pos_snapshot_ts_ + 10'000'000'000LL > timestamp) {
      // write update
      proto_pos->mutable_each_position()->Clear();
      coin::proto::AccountPosition new_pos;
      new_pos.set_exchange(pos.exchange());
      new_pos.set_market_type(pos.market_type());
      for (const auto& ele : pos.each_position()) {
        const auto& symbol = ele.symbol();
        if (curr_pos_map_.count(symbol) &&
            MessageDiff::ApproximatelyEquals(curr_pos_map_[symbol], ele)) {
          continue;
        }
        curr_pos_map_[symbol] = ele;
        new_pos.add_each_position()->CopyFrom(ele);
        has_pos_update = true;
      }
      if (has_pos_update) {
        *proto_pos = new_pos;
      }
    } else {
      // write snapshot
      for (const auto& ele : pos.each_position()) {
        const auto& symbol = ele.symbol();
        curr_pos_map_[symbol] = ele;
      }
      last_log_pos_snapshot_ts_ = timestamp;
      has_pos_update = true;
    }
    return has_pos_update;
  }

 private:
  const coin::proto::AccountRequestProto request_;
  OrderGatewayLog op_log_proto_;
  std::string buf_;
  int64_t raw_msg_id_{0};
  int64_t og_id_{0};
  std::string extension_;
  std::function<void(const coin::proto::OrderGatewayLog& log)> callback_;
  bool as_account_logger_{false};

  // for slim log
  bool use_slim_log_mode_ = true;
  int64_t last_log_pos_snapshot_ts_ = 0;
  int64_t last_log_bal_snapshot_ts_ = 0;
  std::unordered_map<std::string, coin::proto::ProductPosition> curr_pos_map_;
  std::unordered_map<std::string, coin::proto::CurrencyBalance> curr_bal_map_;
};

class RawOrderLogReader : public orio::io::Reader<orio::record::TopicRecord> {
 public:
  RawOrderLogReader(
      const coin::proto::AccountRequestProto& request,
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp,
      size_t max_record_size) {
    std::vector<std::unique_ptr<orio::io::Reader<orio::record::TopicRecord>>> readers;
    std::vector<orio::record::TopicRecord*> buffers;

    for (const auto& path :
         GetFileList(request, root_dir, machine, begin_timestamp, end_timestamp)) {
      readers.emplace_back(std::make_unique<orio::record::IntervalTopicReader>(
          std::unique_ptr<orio::io::Reader<orio::record::TopicRecord>>(
              orio::record::TopicReader::FromFile(path, max_record_size)),
          begin_timestamp,
          end_timestamp));

      buffers_.push_back(std::make_unique<char[]>(max_record_size));
      buffers.push_back(reinterpret_cast<orio::record::TopicRecord*>(buffers_.back().get()));
    }

    reader_ =
        std::make_unique<orio::record::MergedTopicReader>(std::move(readers), std::move(buffers));
  }

  bool Read(orio::record::TopicRecord* out) override { return reader_->Read(out); }

  static std::vector<std::experimental::filesystem::path> GetFileList(
      const coin::proto::AccountRequestProto& request,
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp /* exclusive */) {
    std::vector<std::experimental::filesystem::path> files;

    int64_t one_day = 24 * 60 * 60 * 1'000'000'000LL;
    for (int64_t ts = begin_timestamp; ts < end_timestamp + one_day; ts += one_day) {
      std::string date_str(coin2::exchange::base::symbology::ToDateStr(ts));
      auto path = fmt::format("{}/{}/{}/", root_dir, machine, date_str);
      LOG(INFO) << "trying path " << path;
      if (!std::experimental::filesystem::exists(path)) {
        continue;
      }

      for (const auto& entry : std::experimental::filesystem::directory_iterator(path)) {
        std::string filename = entry.path().filename();
        const auto& log_path_component_optional =
            OrderLogPathComponent::GetLogPathComponentFromFilename(filename);
        if (!log_path_component_optional) {
          LOG(WARNING) << "skip since ill-formatted filename " << filename;
          continue;
        }
        const auto& log_path_component = log_path_component_optional.value();
        tm tm_time{};
        if (log_path_component.owner != request.owner()) {
          LOG(WARNING) << "skip since different owner from request proto " << filename;
          continue;
        } else if (log_path_component.market != request.market_type()) {
          LOG(WARNING) << "skip since different market from request proto " << filename;
          continue;
        } else if (log_path_component.exchange != request.exchange()) {
          LOG(WARNING) << "skip since different exchange from request proto " << filename;
          continue;
        } else if (
            log_path_component.log_time_str.size() != 15 ||
            strptime(log_path_component.log_time_str.c_str(), "%Y%m%d-%H%M%S", &tm_time) ==
                nullptr) {
          LOG(WARNING) << "skip since different first log time format(yyyymmdd-hhmmss) "
                       << filename;
          continue;
        } else if (log_path_component.log_timestamp >= end_timestamp) {
          LOG(WARNING) << "skip since file's log timestamp " << log_path_component.log_timestamp
                       << " is greater than config's end timestamp " << end_timestamp
                       << " file=" << filename;
          continue;
        }
        files.emplace_back(entry.path().string());
      }
    }
    return files;
  }

 private:
  std::unique_ptr<orio::io::Reader<orio::record::TopicRecord>> reader_;
  std::vector<std::unique_ptr<char[]>> buffers_;
};

class OrderLogReader : public orio::io::Reader<OrderGatewayLog> {
 public:
  OrderLogReader(
      const coin::proto::AccountRequestProto& request,
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp,
      size_t record_max_size)
      : reader_(std::make_unique<RawOrderLogReader>(
            request,
            root_dir,
            machine,
            begin_timestamp,
            end_timestamp,
            record_max_size)),
        buf_(new char[record_max_size]) {}

  bool Read(OrderGatewayLog* out) override {
    if (!reader_->Read(reinterpret_cast<orio::record::TopicRecord*>(buf_.get()))) {
      return false;
    }
    auto* r = reinterpret_cast<orio::record::TopicRecord*>(buf_.get());
    if (!out->ParseFromArray(r->data, r->data_len)) {
      throw std::runtime_error("[OrderLogReader] fails to parse proto");
    }
    return true;
  }

  bool Read(OrderGatewayLog* out, int64_t* timestamp) {
    if (!reader_->Read(reinterpret_cast<orio::record::TopicRecord*>(buf_.get()))) {
      return false;
    }
    auto* r = reinterpret_cast<orio::record::TopicRecord*>(buf_.get());
    if (!out->ParseFromArray(r->data, r->data_len)) {
      throw std::runtime_error("[OrderLogReader] fails to parse proto");
    }
    *timestamp = r->timestamp;
    return true;
  }

 private:
  std::unique_ptr<orio::io::Reader<orio::record::TopicRecord>> reader_;
  std::unique_ptr<char[]> buf_;
};

}  // namespace impl

using impl::OrderLogger;
using impl::OrderLogReader;
using impl::RawOrderLogReader;

}  // namespace coin2::exchange::base::order_util
